package com.chukun.flink.stream.window.source;

import com.chukun.flink.stream.util.TimeUtils;
import com.chukun.flink.stream.window.process.bean.EventBean;
import com.chukun.flink.stream.window.process.bean.EventData;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

/**
 * @author chukun
 * @version 1.0.0
 * @description 窗口数据准备
 * @createTime 2022年05月22日 21:48:00
 */
public class SourceWithOutTimestampsWatermarkers implements SourceFunction<EventBean> {

    private volatile boolean isRunning = true;

    // 发送元素的间隔时间
    private long sleepTime;

    public static final String[] words = new String[] {
            "flink","flink","flink",
            "java","hadoop","spark",
            "java","spark","doris"
    };

    public SourceWithOutTimestampsWatermarkers(long sleepTime) {
        this.sleepTime = sleepTime;
    }

    @Override
    public void run(SourceContext<EventBean> ctx) throws Exception {
        int count = 0;
        while (isRunning) {
            if (count >= 16) {
                isRunning  = false;
            }
            // 遍历数组中的元素
            EventBean bean = EventData.BEANS[count % words.length];
            String time = TimeUtils.getHHmmss(System.currentTimeMillis());
            // 发送元素
            ctx.collect(bean);
            System.out.println("send 元素内容 : [ " +bean+" ] now time: "+time);
            if (!bean.getList().get(0).contains("nosleep")) {
                Thread.sleep(sleepTime);
            }
            count++;
        }
    }

    @Override
    public void cancel() {
       this.isRunning = false;
    }
}
